package com.consumerzk.consumerzk.discovery;

import com.common.test.coderesolver.RpcDecoder;
import com.common.test.coderesolver.RpcEncoder;
import com.common.test.entity.RpcRequest;
import com.common.test.serializer.JSONSerializer;
import com.common.test.zk.RegistryCenterConfig;
import com.consumerzk.consumerzk.rpc.RpcConsumerZk;
import com.consumerzk.consumerzk.rpc.UserClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.ArrayList;
import java.util.List;

public class DiscoveryServiceImpl implements DiscoveryService {


    private static CuratorFramework curatorFramework;

    public static List<String> repos;


    public DiscoveryServiceImpl() {
        curatorFramework = CuratorFrameworkFactory.builder()
                .connectString(RegistryCenterConfig.CONNECTING_STR)
                .sessionTimeoutMs(RegistryCenterConfig.SESSION_TIMEOUT)
                .retryPolicy(new ExponentialBackoffRetry(1000, 10)).build();
        curatorFramework.start();
    }






    @Override
    public void discover(String serviceName,RpcConsumerZk rpcConsumer) {

        //获取/rpcNode/下所有协议地址
        String nodePath = RegistryCenterConfig.NAMESPACE+"/"+serviceName;
        try {
             // 初始化链接
            List<String> strings = curatorFramework.getChildren().forPath(nodePath);
            List<UserClientHandler> list = findList(strings);
            rpcConsumer.setUserClientHandlers(list);
        } catch (Exception e) {
            throw new RuntimeException("服务发现获取子节点异常！",e);
        }
        registerWatcher(nodePath,rpcConsumer);

    }

    private void registerWatcher(String path,RpcConsumerZk rpcConsumer){
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework,path,true);
        PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                 PathChildrenCacheEvent.Type type = event.getType();
                List<String> strings = curatorFramework.getChildren().forPath(path);
                if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(type)) {
                    //监听添加
                    System.out.println("add");
                     List<UserClientHandler> list = findList(strings);
                     rpcConsumer.setUserClientHandlers(list);
                }else if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(type)){
                    //添加删除
                    System.out.println("delete");
                    List<UserClientHandler> list = findList(strings);
                    rpcConsumer.setUserClientHandlers(list);
                }
            }
        };
        pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
        try {
            pathChildrenCache.start();
        } catch (Exception e) {
            throw new RuntimeException("监听节点变化异常！",e);
        }
    }
    //2.初始化netty客户端
    private UserClientHandler initClient(String ip, Integer port) throws InterruptedException {
        UserClientHandler userClientHandler =new UserClientHandler();
        EventLoopGroup group = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast( new RpcDecoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(userClientHandler);
                    }
                });

        bootstrap.connect(ip,port).sync();
        userClientHandler.setServerIP(ip);
        userClientHandler.setServerPort(port.toString());
        return userClientHandler;

    }


    private List<UserClientHandler> findList( List<String> strings){
        List<UserClientHandler> userClientHandlers =new ArrayList<>();
        for (String string : strings) {
            String[] split = string.split(":");
            String ip = split[0];
            String port = split[1];
            UserClientHandler userClientHandler;
            try {
                userClientHandler = initClient(ip, Integer.valueOf(port));
                userClientHandlers.add(userClientHandler);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return userClientHandlers;
    }

}
